package com.cl.spark.nodenew;

import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.cl.spark.base.BaseSparkNode;
import com.cl.spark.dto.SparkParam;
import com.cl.spark.dto.SparkResult;
import com.cl.spark.enums.SparkNodeEnum;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class ContainsMatchNode extends BaseSparkNode {

    @Override
    public SparkResult process(SparkParam sparkParam) {
        return SparkResult.success(containsMatch(sparkParam));
    }

    public static JSONArray containsMatch(SparkParam sparkParam) {
        JSONArray res = new JSONArray();

        List<JSONArray> dataList = sparkParam.getDataList();
        JSONObject expression = sparkParam.getNodeExpression();

        JSONArray data1 = dataList.get(0);
        JSONArray data2 = dataList.get(1);

        JSONArray fields1 = expression.getJSONArray("fields1");
        JSONArray fields2 = expression.getJSONArray("fields2");

        for (int i = 0; i < data1.size(); i++) {
            JSONObject item1 = data1.getJSONObject(i);
            boolean match = true;
            for (int j = 0; j < data2.size() && match; j++) {
                JSONObject item2 = data2.getJSONObject(i);
                for (int k = 0; k < fields1.size(); k++) {
                    String value1 = String.valueOf(item1.get(fields1.getString(k)));
                    String value2 = String.valueOf(item2.get(fields2.getString(k)));
                    if (value1 == null || !value1.contains(value2)) {
                        match = false;
                        break;
                    }
                }
            }
            item1.put("MATCH", match);
            res.add(item1);

        }
        return res;
    }


    @Override
    public SparkNodeEnum getType() {
        return SparkNodeEnum.MATCH;
    }
}
